package com.bigdata.quality.config.converter;

import java.util.HashMap;
import java.util.Map;

public class OBMySQLColumnTypeConverter implements ColumnTypeConverter {
  public static final OBMySQLColumnTypeConverter instance = new OBMySQLColumnTypeConverter();

  @Override
  public Map<String, String> toFlinkColumnTypeMap() {
    Map<String, String> columnTypeMap = new HashMap<>();

    // 添加所有映射关系
    columnTypeMap.put("double unsigned zerofill", "double");
    columnTypeMap.put("double precision", "double");
    columnTypeMap.put("double precision unsigned", "double");
    columnTypeMap.put("double precision unsigned zerofill", "double");
    columnTypeMap.put("numeric", "decimal(38,2)");
    columnTypeMap.put("decimal", "decimal(38,2)");
    columnTypeMap.put("boolean", "boolean");
    columnTypeMap.put("date", "date");
    columnTypeMap.put("time", "time(6)");
    columnTypeMap.put("timestamp", "timestamp(6)");
    columnTypeMap.put("datetime", "timestamp(6)");
    columnTypeMap.put("char", "string");
    columnTypeMap.put("varchar", "string");
    columnTypeMap.put("bit", "bytes");
    columnTypeMap.put("binary", "bytes");
    columnTypeMap.put("varbinary", "bytes");
    columnTypeMap.put("tinytext", "string");
    columnTypeMap.put("text", "string");
    columnTypeMap.put("mediumtext", "string");
    columnTypeMap.put("longtext", "string");
    columnTypeMap.put("tinyblob", "bytes");
    columnTypeMap.put("blob", "bytes");
    columnTypeMap.put("mediumblob", "bytes");
    columnTypeMap.put("longblob", "bytes");
    columnTypeMap.put("year", "smallint");
    columnTypeMap.put("enum", "string");
    columnTypeMap.put("json", "string");
    columnTypeMap.put("set", "string");
    columnTypeMap.put("geometry", "bytes");
    columnTypeMap.put("point", "bytes");
    columnTypeMap.put("linestring", "bytes");
    columnTypeMap.put("polygon", "bytes");
    columnTypeMap.put("multipoint", "bytes");
    columnTypeMap.put("multilinestring", "bytes");
    columnTypeMap.put("multipolygon", "bytes");
    columnTypeMap.put("geometrycollection", "bytes");
    columnTypeMap.put("geomcollection", "bytes");
    columnTypeMap.put("tinyint", "tinyint");
    columnTypeMap.put("smallint", "smallint");
    columnTypeMap.put("tinyint unsigned", "smallint");
    columnTypeMap.put("tinyint unsigned zerofill", "smallint");
    columnTypeMap.put("int", "int");
    columnTypeMap.put("mediumint", "int");
    columnTypeMap.put("smallint unsigned", "int");
    columnTypeMap.put("smallint unsigned zerofill", "int");
    columnTypeMap.put("bigint", "bigint");
    columnTypeMap.put("int unsigned", "bigint");
    columnTypeMap.put("int unsigned zerofill", "bigint");
    columnTypeMap.put("mediumint unsigned", "int");
    columnTypeMap.put("mediumint unsigned zerofill", "int");
    columnTypeMap.put("bigint unsigned", "decimal(20,0)");
    columnTypeMap.put("bigint unsigned zerofill", "decimal(20,0)");
    columnTypeMap.put("serial", "decimal(20,0)");
    columnTypeMap.put("float", "float");
    columnTypeMap.put("float unsigned", "float");
    columnTypeMap.put("float unsigned zerofill", "float");
    columnTypeMap.put("real", "double");
    columnTypeMap.put("real unsigned", "double");
    columnTypeMap.put("real unsigned zerofill", "double");
    columnTypeMap.put("double", "double");
    columnTypeMap.put("double unsigned", "double");

    return columnTypeMap;
  }

  @Override
  public String convertColumnType(Map<String, String> flinkColumnTypeMap, String sourceColumnType) {
    return ColumnTypeConverter.super.convertColumnType(flinkColumnTypeMap, sourceColumnType);
  }
}
